package com.zhao.biz.realtime

import java.text.SimpleDateFormat

import com.zhao.common.CommonData
import com.zhao.utils.{CommonUtils, JedisClusterUtil}
import kafka.serializer.StringDecoder
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.JedisCluster

/**
 * Description: 实时统计(根据游戏产生的日志信息,追踪出哪些开挂的游戏玩家),使用广播变量进行优化<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/17 10:52
 *
 * @author 柒柒
 * @version : 1.0
 */

object HotBloodRealTime {
  def main(args: Array[String]): Unit = {

    //步骤:
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))

    //将实例FastDateFormat封装到广播变量中
    val fdf: FastDateFormat = FastDateFormat.getInstance(CommonUtils.getPropertiesValueBykey(CommonData.TIME_PATTERN))
    val bcFdf: Broadcast[FastDateFormat] = sc.broadcast(fdf)
    //采用direct方式从kafka中实时读取数据
    //序列化:将内存中的实例(数据)输出到网络上或是指定的存储介质上存储起来的过程
    //反序列化:从网络上去读数据到内存中,或者从存储介质中读取数据到内存中
    val dsFromKafka: DStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc,
      Map("metadata.broker.list" -> "NODE01:9092,NODE02:9092,NODE03:9092"),
      Set("realTimeGameLog")
    ).map(_._2)

    //实时筛选出"吃疗伤药"的玩家信息,并抽离出关注的信息,用户名,时间
    //1	0	0	10	3	324	325	风道	裁决之杖	96609	1	装备回收	2018-02-01 11:08:07
    val filterAndStandardDS: DStream[(String, Long)] = dsFromKafka.filter(perMsg => {
      val arr = perMsg.split("\t")
      //时间类型是:11  -->索引号是:3
      val eventType = arr(3).trim.toInt
      //道具的名称是:疗伤药 ->索引号是:8
      val item = arr(8).trim
      //返回
      eventType == 11 && "疗伤药".equals(item)
    }).map(perMsg => {
      val arr = perMsg.split("\t")
      //玩家名
      val username = arr(7).trim
      //玩游戏动作触发执行的时间
      val time = arr(12).trim
      //返回时间格式:2018-02-01 11:08:07
      (username, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(time).getTime)
    })

    //根据业务剥离出开了外挂的玩家
    val groupAfterDS: DStream[(String, Iterable[Long])] = filterAndStandardDS.groupByKeyAndWindow(Seconds(10), Seconds(5))

    //a)每隔5秒,统计过去10秒内吃疗伤药的次数>=5 初次筛选(疑似开了挂的玩家)
    val blackListPlayer: DStream[(String, Double)] = groupAfterDS.filter(_._2.size >= 5)
      //b)平均每次吃疗伤药的时间间隔<1s
      .map(perEle => {
        //将当前玩家吃药的所有时间进行排序
        val sortedList: List[Long] = perEle._2.toList.sortWith(_ > _)

        //求出最后一次吃药的时间
        val lastTime = sortedList(0)

        //求出第一次吃药的时间
        val firstTime = sortedList(sortedList.size - 1)

        //求平均时间
        val avgTime = (lastTime - firstTime) * 1.0 / sortedList.size

        //返回
        (perEle._1, avgTime)
      }).filter(_._2 < 1 * 1000)

    //将开了挂的玩家信息保存在redis中
    //将开了挂的玩家名,以及每次吃疗伤药的平均时间落地到db中
    //foreachRDD,DStream中只包含一个RDD,不是多个,可以与其他RDD的其他算子mapPartitions,foreachPartitions进行类比
    blackListPlayer.foreachRDD(rdd =>{
      if (!rdd.isEmpty()) {
        rdd.foreachPartition(itr =>{
          if (!itr.isEmpty){
            //获得JedisCluster的实例
            val jedisCluster: JedisCluster = JedisClusterUtil.getJedisClusterInstanceFromPool
            //设想:illegalUsers 张三 1.24 李四 2.45
            //redis:读->11万次每秒 写:8.1万次每秒
            itr.foreach(perEle =>{
              jedisCluster.hset("illegalUsers",perEle._1,perEle._2.toString)
            })
            //释放
            JedisClusterUtil.resourceRelease(jedisCluster)
          }
        })
      }
    })
    //启动
    ssc.start()
    //等待结束
    ssc.awaitTermination()
  }
}























